package fox.framework.rpc.client;

import fox.framework.rpc.client.message.RpcClientRequestInterceptor;
import fox.framework.rpc.client.node.RpcMetadata;
import fox.framework.rpc.client.node.RpcNode;
import fox.framework.rpc.client.node.event.NodeChannelCloseListener;
import fox.framework.rpc.client.node.rule.Rule;
import fox.framework.rpc.context.definition.RpcDefinition;
import fox.framework.rpc.context.order.Order;
import fox.framework.rpc.converter.RpcMessageConverter;
import fox.framework.rpc.endpoint.channel.RpcChannel;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

/**
 * Rpc客户端实际入口类
 *
 * @author cuichao
 * @Description: AbstractRpcClient
 * @Date: create in 2021/1/6 16:12
 */
public abstract class AbstractRpcClient extends ConfigurableRpcClient implements NodeChannelCloseListener {

    public AbstractRpcClient(String bind, int port) {
        super(bind, port);
    }


    /**
     * Rpc服务元数据空间
     */
    private final ConcurrentHashMap<String, RpcMetadata> metadataMap = new ConcurrentHashMap<>(128);

    /**
     * Rpc通讯节点
     */
    private final ConcurrentMap<String, RpcNode> nodeCache = new ConcurrentHashMap<>(256);
    /**
     * 节点地址 元空间数据
     */

    private final ConcurrentHashMap<String, Set<RpcMetadata>> nodeMetaRelation = new ConcurrentHashMap<>(256);
    /**
     * 消息转换器映射缓存
     */
    private final ConcurrentMap<String, RpcMessageConverter> convertersCache = new ConcurrentHashMap<>(32);
    /**
     * 客户端请求拦截执行器
     */
    private List<RpcClientRequestInterceptor> registeredRequestInterceptor = new ArrayList<>();
    /**
     * 需要注册的本地服务节点
     */
    private final List<String> registerNodes = new ArrayList<>();


    /**
     * 根据RpcDefinition 构建一个metaData
     *
     * @param definition
     * @return
     */
    protected abstract RpcMetadata buildMetadata(RpcDefinition definition);

    /**
     * 创建一个Rpc节点
     *
     * @param metadata
     * @param nodeAddress
     * @return
     */
    protected abstract RpcNode doCreateRpcNode(RpcMetadata metadata, String nodeAddress);

    /**
     * 创建RpcNode
     *
     * @param metadata
     * @param nodeAddress
     * @return
     */
    private RpcNode createRpcNode(RpcMetadata metadata, String nodeAddress) {
        RpcNode node = doCreateRpcNode(metadata, nodeAddress);
        nodeCache.put(node.getAddress(), node);

        //查看RpcNode 和 MetaData缓存中是否有值 没有则添加
        if(!nodeMetaRelation.containsKey(node.getAddress())){
            //防止同步问题
            nodeMetaRelation.putIfAbsent(node.getAddress(),new HashSet<>());
        }
        Set<RpcMetadata> relation = nodeMetaRelation.get(node.getAddress());
        relation.add(metadata);
        return node;
    }

    @Override
    public void channelCloseListener(RpcNode node) {
        doUpdateNode(node, RpcNode.RPC_NODE_PAUSE);
    }

    @Override
    public void updateNode(String metaspace, String node, int state){
        RpcMetadata meta = metadataMap.get(metaspace);
        if(meta == null){
            return;
        }
        //从缓存中获取如果没有则表示没创建过
        RpcNode rpcNode = nodeCache.getOrDefault(node, null);
        if(rpcNode == null){
            rpcNode = createRpcNode(meta,node);
        }
        doUpdateNode(rpcNode,state);
    }

    /**
     * 更新节点状态
     *
     * @param node
     * @param status
     */
    private void doUpdateNode(RpcNode node, int status){
        Set<RpcMetadata> metaSet = nodeMetaRelation.get(node.getAddress());
        for (RpcMetadata metadata : metaSet) {
            if(metadata.containNode(node)){
                metadata.setNodeState(node,status);
            }else {
                node.setState(status);
                metadata.addNode(node);
            }
        }
    }
    /**
     * 根据元数据空间名获取到Rpc通讯节点
     *
     * @param metadata
     * @return
     */
    protected RpcNode chooseNode(RpcMetadata metadata) {
        String metaspace = metadata.metaspace();
        //懒加载且也未被抓取过元数据 #sync
        //  (nodes.isEmpty() && isAllowFetchIfListEmpty() 是否允许没有可用的RpcNode之后再次向注册中心强制抓取策略暂忽略
        if (metadata.isLazy() && !metadata.isFetched()) {
            //dcl获取每个不用metadata的锁实现同步防止重复调用
            if (metadata.isLazy() && !metadata.isFetched()) {
                fetchNodes(Collections.singletonList(metaspace));
            }
        }
        //进行choose
        Rule rule = metadata.rule();
        return rule.choose(metadata);
    }

    /**
     * 节点进行远程连接
     *
     * @param node
     */
    protected void connectNode(RpcNode node) {
        RpcChannel channel = getEndpoint().connection(node);
        node.setChannel(channel);
        doUpdateNode(node, RpcNode.RPC_NODE_RUNNING);
    }
    /**
     * 根据MediaType获取转换器
     *
     * @param mediaType
     * @return
     */
    protected RpcMessageConverter getSupportConverter(String protocol, String mediaType) {
        String key = protocol + ":" + mediaType;
        RpcMessageConverter converter = convertersCache.getOrDefault(key, null);
        if (converter != null) {
            return converter;
        }
        converter = getMessageConverters().getSupportConverter(protocol, mediaType);
        convertersCache.put(key, converter);
        return converter;
    }

    /**
     * 根据转换器名称获取转换器
     *
     * @param name
     * @return
     */
    protected RpcMessageConverter getConverterByName(String name) {
        return getMessageConverters().get(name);
    }

    /**
     * 获取全部的MetaData
     *
     * @return
     */
    protected ConcurrentHashMap<String, RpcMetadata> getMetaMap() {
        return metadataMap;
    }

    public void run() throws Exception {
        //刷新前执行部分操作
        prepareRefresh();
        //解析rpc注册表定义
        refreshMetaspaceInfo();
        //启动endpoint
        enableEndPoint();
        //启动注册器
        enableRegister();
        //注册本机的服务节点
        registerProviders(getProviderMetaspaces());
        //获取非懒加载的RpcNode
        fetchNodes(getNoLazyMetaspace());
    }

    /**
     * 刷新前操作
     */
    protected void prepareRefresh() {
        clearCache();
        //注册客户端请求拦截器
        registerRpcRequestInterceptor(getUnRegisteredInterceptors());
    }

    /**
     * 根据RpcDefinition构建逻辑空间列表
     */
    private void refreshMetaspaceInfo() {
        metadataMap.clear();
        registerNodes.clear();
        for (RpcDefinition definition : getDefinitions()) {
            if (definition.isProvider()) {
                registerNodes.add(definition.metaspace());
            } else {
                RpcMetadata metadata = buildMetadata(definition);
                metadataMap.put(metadata.metaspace(), metadata);
            }
        }
    }


    /**
     * 获取非懒加载的源数据空间
     *
     * @return
     */
    private List<String> getNoLazyMetaspace() {
        return metadataMap.entrySet().stream()
                .filter(v -> !v.getValue().isLazy()).map(v -> v.getKey())
                .collect(Collectors.toList());
    }


    private List<String> getProviderMetaspaces() {
        return registerNodes;
    }

    private void clearCache(){
        metadataMap.clear();
        nodeCache.clear();
        nodeMetaRelation.clear();
        convertersCache.clear();
        registerNodes.clear();
    }


    protected void registerRpcRequestInterceptor(List<RpcClientRequestInterceptor> interceptors) {
        List<RpcClientRequestInterceptor> orderInterceptors = new ArrayList<>();
        List<RpcClientRequestInterceptor> nonOrderInterceptors = new ArrayList<>();
        for (RpcClientRequestInterceptor interceptor : interceptors) {
            if (interceptor instanceof Order) {
                orderInterceptors.add(interceptor);
            } else {
                nonOrderInterceptors.add(interceptor);
            }
        }
        sortByOrderInterceptors(orderInterceptors);
        registeredRequestInterceptor.clear();
        registeredRequestInterceptor.addAll(orderInterceptors);
        registeredRequestInterceptor.addAll(nonOrderInterceptors);
    }


    protected List<RpcClientRequestInterceptor> getRpcRequestInterceptor() {
        return this.registeredRequestInterceptor;
    }

    private void sortByOrderInterceptors(List<RpcClientRequestInterceptor> interceptors) {
       Order.sort(interceptors);
    }
}
